package cn.doitedu.dynamic_rule.demos.timer;

import cn.doitedu.dynamic_rule.functions.SourceFunctions;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by yuanhailong on 2021/8/24.
 * @desc
 *   定时器使用的主程序
 *
 *
 *   发送数据命令
 *   bin/kafka-console-producer.sh --broker-list 192.168.112.101:9092 --topic timerhl
 *
 */
public class TimerApplicationDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.enableCheckpointing(60000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(100000);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(100);
        //env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
        //env.setRestartStrategy(RestartStrategies.failureRateRestart(3, org.apache.flink.api.common.time.Time.seconds(300), org.apache.flink.api.common.time.Time.seconds(10)));

        //DataStreamSource<String> socketTextStream = env.socketTextStream("192.168.112.101", 7777);

        DataStreamSource<String> kafkaStream = env.addSource(SourceFunctions.getKafkaEventSource("timerhl"));

        SingleOutputStreamOperator<String>  dataStream= kafkaStream.filter(str -> StringUtils.isNotEmpty(str) && str.split(",").length >= 3)
                .map(str -> new SensorReading(str.split(",")[0].trim(), Long.valueOf(str.split(",")[1].trim()), Double.valueOf(str.trim().split(",")[2].trim())))
                .keyBy(sensorReading -> sensorReading.getSensorName())
                .process(new SensorKeyedProcessFunction());

        dataStream.print();

        env.execute("timerserver");

    }

}
